package com.hopu.user.time;

import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.annotation.JsonIdentityInfo;
import com.hopu.bean.time.TimeTotal;
import com.hopu.util.HBaseUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.io.IOException;
import java.sql.Time;
import java.util.ArrayList;
import java.util.List;

public class TimeTotalAna {

    public static void main(String[] args) {
        //加载日志文件，生成RDD
        SparkConf conf = new SparkConf().setAppName("").setMaster("local");
        JavaSparkContext context = new JavaSparkContext(conf);
        JavaRDD<String> rdd = context.textFile("D:/word/user_session.log");

        //把原始日志数据转换成实体类的RDD
        JavaPairRDD<TimeTotal, Integer> pairRDD = rdd.mapToPair(t -> {
            JSONObject json = (JSONObject) JSONObject.parse(t);
            String hour = json.getJSONObject("odate").getString("hour");
            String type = json.getString("event_type");

            return new Tuple2<>(new TimeTotal(hour, type), 1);
        });

        ArrayList<TimeTotal> list = new ArrayList<>();

        //过滤时间
        for (int i = 0; i < 24; i++) {
            TimeTotal transform = transform(pairRDD, i + "");
            list.add(transform);
        }
        calAndPut(list);

    }

    public static TimeTotal transform(JavaPairRDD<TimeTotal, Integer> pairRDD, String hour) {
        //过滤时间段
        JavaPairRDD<TimeTotal, Integer> hourRDD = pairRDD.filter(t -> t._1.getHour().equals(hour));
        /*
        (TimeTotal(hour=0, type=parchase, viewCount=0, cartCount=0, parchaseCount=0, totalCount=0, viewPercent=null, cartPercent=null, parchasePercent=null, totalPercent=null),1)
        (TimeTotal(hour=0, type=cart, viewCount=0, cartCount=0, parchaseCount=0, totalCount=0, viewPercent=null, cartPercent=null, parchasePercent=null, totalPercent=null),1)
         */
        //
        JavaPairRDD<TimeTotal, Iterable<Integer>> groupByType = hourRDD.groupByKey();

        JavaRDD<TimeTotal> map = groupByType.map(t -> {
            TimeTotal result = t._1;
            int count = 0;
            for (int v : t._2) {
                count += v;
            }
            if (result.getType().equals("view")) {
                result.setViewCount(count);
            } else if (result.getType().equals("cart")) {
                result.setCartCount(count);
            } else if (result.getType().equals("parchase")) {
                result.setParchaseCount(count);
            } else {
                throw new Exception("类型错误");
            }
            return result;
        });

//        map.foreach(t -> System.out.println(t));
        List<TimeTotal> take = map.take(3);

        return getResult(take);
    }

    public static TimeTotal getResult(List<TimeTotal> take) {
        TimeTotal result = take.get(0);
        if (result.getType().equals("view")) {
            result.setCartCount(take.get(1).getCartCount());
            result.setParchaseCount(take.get(2).getParchaseCount());
        } else if (result.getType().equals("parchase")) {
            result.setViewCount(take.get(1).getViewCount());
            result.setCartCount(take.get(2).getCartCount());
        } else if (result.getType().equals("cart")) {
            result.setParchaseCount(take.get(1).getParchaseCount());
            result.setViewCount(take.get(2).getViewCount());
        }
        result.setTotalCount(result.getViewCount()+result.getCartCount()+result.getParchaseCount());

        System.out.println(result);

        return result;
    }

    public static void calAndPut(ArrayList<TimeTotal> list) {
        int totalViewCount = 0;
        int totalCartCount = 0;
        int totalParchaseCount = 0;
        int total = 0;

        //统计总量
        for (TimeTotal timeTotal: list) {
            totalViewCount += timeTotal.getViewCount();
            totalCartCount += timeTotal.getCartCount();
            totalParchaseCount += timeTotal.getParchaseCount();
            total += timeTotal.getTotalCount();
        }

        //计算百分比，并且写入到HBase
        for (TimeTotal timeTotal: list) {
            //计算百分比
            String viewPercent = (float)timeTotal.getViewCount()*100 / totalViewCount + "%";
            String cartPercent = (float)timeTotal.getCartCount()*100 / totalCartCount + "%";
            String parchasePercent = (float)timeTotal.getParchaseCount()*100 / totalParchaseCount + "%";
            String totalPercent = (float)timeTotal.getTotalCount()*100 / total + "%";
            System.out.println(timeTotal.getHour() + "\t"  + viewPercent + "\t" + cartPercent + "\t" + parchasePercent
            + "\t" + totalPercent);

            //写入到HBase
            String[] columns = {"hour", "viewCount", "cartCount", "parchaseCount", "totalCount", "viewPercent",
                    "cartPercent", "parchasePercent", "totalPercent"};
            String[] values = {timeTotal.getHour(), timeTotal.getViewCount()+"",
            timeTotal.getCartCount()+"", timeTotal.getParchaseCount()+"", timeTotal.getTotalCount()+"",
            viewPercent, cartPercent, parchasePercent, totalPercent};

            try {
                HBaseUtils.putOneRowToHbase("time", timeTotal.getHour(), "totalinfo", columns, values);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

    }
}
